package cn.ac.iscas.service.flink;


import cn.ac.iscas.config.FlinkConfig;
import cn.ac.iscas.service.executor.flink.FlinkJobExecutor;
import com.iscas.datasong.lib.common.DataSongException;
import com.iscas.datasong.lib.util.DataSongJsonUtils;
import com.iscas.datasong.lib.util.OkHttpClientUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Service
public class LogService {
    @Autowired
    private FlinkConfig flinkConfig;

    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @Autowired
    private FlinkJobExecutor flinkJobExecutor;

    //格式：Map<igJobId, flinkJobId>
    public static Map<String, String> runningJobIds = new ConcurrentHashMap<>();

    //格式：Map<String jobId, Map<String subtaskId, String subtaskStatus>>
    private Map<String, Map<String, String>> latestAllJobStatus = new ConcurrentHashMap<>();

    public String getJobLog(String jobId, String trainId) {
        Map<String, String> lastestJobStatus = null;
        if (latestAllJobStatus.containsKey(jobId)) {
            lastestJobStatus = latestAllJobStatus.get(jobId);
        }

        List<String> jobLogs = new ArrayList<>();
        String baseUrl = new StringBuffer("http://")
                .append(flinkConfig.getFlinkJobmanagerAddr())
                .append(":")
                .append(flinkConfig.getFlinkHistorymanagerPort())
                .append("/jobs/")
                .append(jobId)
                .toString();
        OkHttpClientUtils httpClient = new OkHttpClientUtils();
        String response = null;
        try {
            response = httpClient.doGet(baseUrl);
            if (response.contains("errors")) {
                baseUrl = baseUrl.replace(":8082", ":8081");
            }
            response = httpClient.doGet(baseUrl);
            Map<String, Object> nowJobStatus = DataSongJsonUtils.fromJson(response, Map.class);
            String nowJobState = nowJobStatus.getOrDefault("state", "").toString();

            //如果job当前状态为正在进行时状态（包括），则需要动态更新日志
            if (nowJobState.equals("RUNNING") || nowJobState.equals("FAILING")
                    || nowJobState.equals("CANCELLING") || nowJobState.equals("RESTARTING")
                    || nowJobState.equals("RECONCILING") || nowJobState.equals("CREATED")) {


                List<Map<String, Object>> vertices = (ArrayList)nowJobStatus.get("vertices");
                Map<String, String> nowStatusMap = new HashMap<>();
                for (Map<String, Object> vertice : vertices) {
                    String subtaskId = (String) vertice.get("id");
                    String nowSubtaskStatus = (String)vertice.get("status");
                    String subtaskName = (String)vertice.get("name");
                    String subtaskStartTime = DateTimeFormatter.ofPattern("YYYY-MM-dd HH:mm:ss").format(LocalDateTime.ofInstant(Instant.ofEpochMilli((Long)vertice.get("start-time")), ZoneId.systemDefault()));
                    String subtaskDuration = String.valueOf(vertice.get("duration"));

                    //第一次获取job状态，将当前状态填充到子任务状态map中，同时返回所有的子任务状态
                    if (lastestJobStatus == null) {
                        String subtaskLog = String.format("Subtask (%s) %s start at %s, duration %sms, status: [%s]",
                                subtaskId, subtaskName, subtaskStartTime, subtaskDuration, nowSubtaskStatus);
                        jobLogs.add(subtaskLog);
                        nowStatusMap.put(subtaskId, nowSubtaskStatus);
                        continue;
                    }
                    //如果非第一次获取job状态，则将本次状态与上一次状态进行比较，如果发生改变再输出
                    if (!lastestJobStatus.get(subtaskId).equalsIgnoreCase(nowSubtaskStatus)) {
                        String subtaskLog = String.format("Subtask (%s) %s start at %s, duration %sms, status changed from [%s] to [%s] ",
                                subtaskId, subtaskName, subtaskStartTime, subtaskDuration, lastestJobStatus.get(subtaskId), nowSubtaskStatus);
                        jobLogs.add(subtaskLog);
                        //更新当前subtask状态
                        latestAllJobStatus.get(jobId).put(subtaskId, nowSubtaskStatus);
                    }
                }
                if (lastestJobStatus == null) {
                    latestAllJobStatus.put(jobId, nowStatusMap);
                }
            } else { //如果job当前状态为固定状态，包括，则日志为最终状态，不需要更新
                String jobStartTime = DateTimeFormatter.ofPattern("YYYY-MM-dd HH:mm:ss").format(LocalDateTime.ofInstant(Instant.ofEpochMilli((Long)nowJobStatus.get("start-time")), ZoneId.systemDefault()));
                String jobDuration = String.valueOf(nowJobStatus.get("duration"));

                String jobLog = String.format("Job (%s) start at %s, duration %sms, status：[%s] ",
                        jobId, jobStartTime, jobDuration, nowJobState);
                jobLogs.add(jobLog);
            }
        } catch (IOException | DataSongException e) {
            e.printStackTrace();
        }

        String log = String.join("\n", jobLogs);
        messagingTemplate.convertAndSend("/topic/log/" + trainId, log);

        return log;
    }

    public String sendLog(String flinkJobId, String jobId) {
        Map<String, String> lastestJobStatus = null;
        if (latestAllJobStatus.containsKey(flinkJobId)) {
            lastestJobStatus = latestAllJobStatus.get(flinkJobId);
        }

        List<String> jobLogs = new ArrayList<>();
        String baseUrl = new StringBuffer("http://")
                .append(flinkConfig.getFlinkJobmanagerAddr())
                .append(":")
                .append(flinkConfig.getFlinkHistorymanagerPort())
                .append("/jobs/")
                .append(flinkJobId)
                .toString();
        OkHttpClientUtils httpClient = new OkHttpClientUtils();
        String response = null;
        try {
            response = httpClient.doGet(baseUrl);
            System.out.println("RESSS: " + response);
            if (response.contains("errors")) {
                baseUrl = baseUrl.replace(":8082", ":8081");
            }
            System.out.printf("baseUrl: " + baseUrl);
            response = httpClient.doGet(baseUrl);
            if (response.contains("errors")) {
                return null;
            }
            Map<String, Object> nowJobStatus = DataSongJsonUtils.fromJson(response, Map.class);
            String nowJobState = nowJobStatus.getOrDefault("state", "").toString();

            //如果job当前状态为正在进行时状态（包括），则需要动态更新日志
            if (nowJobState.equals("RUNNING") || nowJobState.equals("FAILING")
                    || nowJobState.equals("CANCELLING") || nowJobState.equals("RESTARTING")
                    || nowJobState.equals("RECONCILING") || nowJobState.equals("CREATED")) {
                List<Map<String, Object>> vertices = (ArrayList)nowJobStatus.get("vertices");
                Map<String, String> nowStatusMap = new HashMap<>();
                for (Map<String, Object> vertice : vertices) {
                    String subtaskId = (String) vertice.get("id");
                    String nowSubtaskStatus = (String)vertice.get("status");
                    String subtaskName = (String)vertice.get("name");
                    String subtaskStartTime = DateTimeFormatter.ofPattern("YYYY-MM-dd HH:mm:ss").format(LocalDateTime.ofInstant(Instant.ofEpochMilli((Long)vertice.get("start-time")), ZoneId.systemDefault()));
                    String subtaskDuration = String.valueOf(vertice.get("duration"));

                    //第一次获取job状态，将当前状态填充到子任务状态map中，同时返回所有的子任务状态
                    if (lastestJobStatus == null) {
                        String subtaskLog = String.format("Subtask (%s) %s start at %s, duration %sms, status: [%s]",
                                subtaskId, subtaskName, subtaskStartTime, subtaskDuration, nowSubtaskStatus);
                        jobLogs.add(subtaskLog);
                        nowStatusMap.put(subtaskId, nowSubtaskStatus);
                        continue;
                    }
                    //如果非第一次获取job状态，则将本次状态与上一次状态进行比较，如果发生改变再输出
                    if (!lastestJobStatus.get(subtaskId).equalsIgnoreCase(nowSubtaskStatus)) {
                        String subtaskLog = String.format("Subtask (%s) %s start at %s, duration %sms, status changed from [%s] to [%s] ",
                                subtaskId, subtaskName, subtaskStartTime, subtaskDuration, lastestJobStatus.get(subtaskId), nowSubtaskStatus);
                        jobLogs.add(subtaskLog);
                        //更新当前subtask状态
                        latestAllJobStatus.get(flinkJobId).put(subtaskId, nowSubtaskStatus);
                    }
                }
                if (lastestJobStatus == null) {
                    latestAllJobStatus.put(flinkJobId, nowStatusMap);
                }
            } else { //如果job当前状态为固定状态，包括，则日志为最终状态，不需要更新
                String jobStartTime = DateTimeFormatter.ofPattern("YYYY-MM-dd HH:mm:ss").format(LocalDateTime.ofInstant(Instant.ofEpochMilli((Long)nowJobStatus.get("start-time")), ZoneId.systemDefault()));
                String jobDuration = String.valueOf(nowJobStatus.get("duration"));

                String jobLog = String.format("Job (%s) start at %s, duration %sms, status：[%s] ",
                        flinkJobId, jobStartTime, jobDuration, nowJobState);
                jobLogs.add(jobLog);
                LogService.runningJobIds.remove(flinkJobId);
                System.out.println(String.format("Flink job [%s] finished!", flinkJobId.trim()));

                flinkJobExecutor.flinkJobFinishedNotice(jobId, flinkJobId, nowJobState, jobDuration);
            }
        } catch (IOException | DataSongException e) {
            e.printStackTrace();
        }

        String log = String.join(System.lineSeparator(), jobLogs) + System.lineSeparator();
        //TODO:前台完成后打开
        messagingTemplate.convertAndSend("/topic/log/"+jobId, log);
        System.out.println("+++" + log);
        return log;
    }

    @Scheduled(fixedRate = 5000)
    public void sendLog() {
        for (Map.Entry<String, String> entry : runningJobIds.entrySet()) {
            sendLog(entry.getKey(), entry.getValue());
        }
    }
}
